package com.xzy.rabbit.demo.delayqueue.c;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.xzy.rabbit.common.util.RabbitUtil;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * 延迟队列——基于MQ插件”rabbitmq-delayed-message-exchange“
 *
 * @author xzy
 * @date 2021/9/29 22:24
 */
class Main {
    /**
     * 发送的消息总数
     */
    private static final Integer MESSAGE_COUNT = 3;
    /**
     * 发送每条消息的延迟时间
     */
    private static final long WORK_TIME = 10000L;

    /**
     * 消息过期时间
     */
    private static final long MSG_DELAY_TIME_1 = 1000L;
    private static final long MSG_DELAY_TIME_2 = 3000L;

    /**
     * 交换机名称
     */
    private static final String EXCHANGE_NAME_DELAY = "exchange.delay.plugin";
    /**
     * 队列名称
     */
    private static final String QUEUE_NAME_DELAY = "queue.delay.plugin";
    /**
     * 路由键
     */
    private static final String ROUTING_KEY_DELAY = "delay";

    /*
     * Producer----k1+DT----[延迟交换机]----k1----[延迟队列]----k1----Consumer
     */

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 1.获取连接
        Connection connection = RabbitUtil.getConnection();

        // 2.准备执行环境
        prepare(connection);

        // 3.发送、接收消息
        Thread producer1s = new Thread(new Producer(connection, EXCHANGE_NAME_DELAY, ROUTING_KEY_DELAY, MESSAGE_COUNT, WORK_TIME, MSG_DELAY_TIME_1), "Producer 1s");
        Thread producer3s = new Thread(new Producer(connection, EXCHANGE_NAME_DELAY, ROUTING_KEY_DELAY, MESSAGE_COUNT, WORK_TIME, MSG_DELAY_TIME_2), "Producer 3s");
        Thread consumer = new Thread(new Consumer(connection, QUEUE_NAME_DELAY), "Consumer");

        // 故意先发送延迟时间长的消息
        producer3s.start();
        producer1s.start();
        consumer.start();

        producer1s.join();
        producer3s.join();
        consumer.join();
    }

    /**
     * 准备执行环境：交换机、队列、绑定
     */
    private static void prepare(Connection connection) throws IOException {
        // 1.获取信道
        Channel channel = connection.createChannel();

        // 2.声明延迟交换机（真正的延迟交换机！！！）
        Map<String, Object> arguments = new HashMap<>(1);
        arguments.put("x-delayed-type", "direct");
        channel.exchangeDeclare(EXCHANGE_NAME_DELAY, "x-delayed-message", true, false, arguments);

        // 3.声明延迟队列
        channel.queueDeclare(QUEUE_NAME_DELAY, false, false, false, null);

        // 4.绑定
        channel.queueBind(QUEUE_NAME_DELAY, EXCHANGE_NAME_DELAY, ROUTING_KEY_DELAY);
    }
}
